Source code for src.cmd.process

import sys, subprocess, time
from os.path import join
import cardsharp as cs
from util import *
from ..util import check_file, RUNDATETIME, get_state_name
from ..configuration import db_info, config

def process(cmd):
    '''Spawns a process subprocess for segments.'''
    args = cmd.args
    #get the segments to run
    _segments = [segment_run_map[seg] for seg in args.segments.split('|')]
    
    regions = cs.load(**{'source': db_info['stand']['name'], 
                         'dataset': 'region', 
                         'format': 'mysql',
                         'user': db_info['stand']['user'], 
                         'pwd': db_info['stand']['pass'], 
                         'load_as_null': ['',]
                         }
                      )
    
    region = args.where if args.where else 'all'
    
    #validate options
    #validate include and exclude rules mutual exclusion
    if args.include_rules and args.exclude_rules:
        print 'Can not specify include and exclude rules. Please choose one or the other.'
        sys.exit()
    
    #validate rule dir
    check_file(args.rules.strip(), as_dir = True)
    
    
    opt = {'out_dir': check_file(args.out_dir.strip(), as_dir = True),
           'meta_dir': check_file(args.metadata.strip(), as_dir = True),
           'rules': '%s/process_rules' % args.rules.strip(), 
           'out_format': args.out_format.strip(),
           'validate_rules': True,
           'in_where':args.where,
           'verbose': args.verbose,
           'db_info': db_info['stand'],
           'halt_on_err': False,
           'limit': args.limit,
           'phase_id':'2',
           'include_rules': args.include_rules,
           'exclude_rules': args.exclude_rules,
          }
    
    for segment in list(_segments): 
        if segment in ('demographic'): #removed 'supervision'
            _log_dir = join(config.log_dir, 'set_%i' % config.current_set, 
                            'process', segment, region, RUNDATETIME)
            _opt = opt.copy()
            _opt['segment'] = segment
            _opt['log_dir'] = create_log_dir(_log_dir)
            cmd.task_queue.put(_opt)
            _segments.remove(segment)
    
    cmd.start() #start the supervision and demographic workers
    
    for segment in _segments:
        if segment == 'supervision': continue #removed supervision from processing
        opt['segment'] = segment
        if not args.regions:
            _log_dir = join(config.log_dir, 'set_%i' % config.current_set, 
                            'process', segment, region, RUNDATETIME)
            opt['log_dir'] = create_log_dir(_log_dir)
        _opt = opt.copy()
        
        #if where is not declared than run all states
        #if parallel specified run all states in parellel    
        if not args.where and args.parallel:
            for row in regions:
                __opt = opt.copy()
                __opt['in_where'] = 'state=%s' % row['id']
                cmd.task_queue.put(__opt)
        
        else:
            if args.regions:
                for region in args.regions:
                    __opt = opt.copy()
                    #log dir not created above, create it
                    _log_dir = join(config.log_dir, 'set_%i' % config.current_set, 
                                    'process', segment, region, RUNDATETIME)
                    __opt['log_dir'] = create_log_dir(_log_dir)
                    __opt['in_where'] = 'state=%s' % get_state_name(region)
                    cmd.task_queue.put(__opt)
            else:
                cmd.task_queue.put(_opt)
        
    cmd.start() #start the arrest and sentence workers
    cmd.task_queue.join() #wait for everything to finish
    cmd.flush() #flush the cmd logs